package com.huan.maket;

import com.huan.bean.ChannelPromotionCount;
import com.huan.bean.MarketingUserBehavior;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;

public class AppMarketingByChannel {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic( TimeCharacteristic.EventTime );
        env.setParallelism( 1 );

        //1.自定义数据源读取数据
        DataStream<MarketingUserBehavior> dataStream = env.addSource( new SimulatedMarketingBehaviorSource() )
                .assignTimestampsAndWatermarks( new AscendingTimestampExtractor<MarketingUserBehavior>() {
                    @Override
                    public long extractAscendingTimestamp(MarketingUserBehavior element) {
                        return element.getTimestamp();
                    }
                } );

        //2.分渠道开窗统计
        SingleOutputStreamOperator<ChannelPromotionCount> resultStream = dataStream.filter( data -> !"UNINSTALL".equals( data.getBehavior() ) ) //过滤掉UNINSTALL
                .keyBy( "channel", "behavior" ) // 分组
                .timeWindow( Time.hours( 1 ), Time.seconds( 5 ) ) //滑窗
                .aggregate( new MarkeingCountAgg(), new MarketingCountResult() );

        resultStream.print();

        env.execute("AppMarketingByChannel");

    }
}
